﻿using System.Collections.Generic;
using System;
using System.Diagnostics;
using System.Linq;
using System.Linq.Expressions;
using System.Net;
using System.Text;
using System.Threading;
using Utility;
using Microsoft.WindowsAzure;
using Microsoft.WindowsAzure.ServiceRuntime;
using Microsoft.WindowsAzure.StorageClient;
using Microsoft.WindowsAzure.Diagnostics;
using System.IO;
using System.Net.Sockets;
using Microsoft.SqlServer.Server;
using System.Data.SqlClient;

namespace WorkerRole
{
    public class WorkerRole : RoleEntryPoint
    {
        string connectionString;
        string instance_id;

        TimeSpan waitingTime = new TimeSpan(0, 0, 0);
        TimeSpan parseTime = new TimeSpan(0, 0, 0);
        TimeSpan taskCreation = new TimeSpan(0, 0, 0);
        TimeSpan processingTime = new TimeSpan(0, 0, 0);
        TimeSpan writingTime = new TimeSpan(0, 0, 0);

        int syncCount = 0;

        public override void Run()
        {
            try
            {
                double[] queueResult = new double[15];
                double[] blobResult = new double[10];
                double[] tableResult = new double[20];
                double[] sqlResult = new double[6];
                int numMessage = 0;
                connectionString = RoleEnvironment.GetConfigurationSettingValue("DataConnectionString");
                QueueHelper QueueObject = new QueueHelper(connectionString);
                CloudQueueMessage msg = new CloudQueueMessage("");
                string resultMsg = string.Empty;
                while (true)
                {
                    /* Check for work in Bench Queue */
                    numMessage = 0;
                    while (numMessage == 0)
                    {
                        QueueObject.GetApproxMessageCount(DataConstants.INPUT_PARAS, out numMessage);
                        if (numMessage == 0) Thread.Sleep(5000);
                    }
                    QueueObject.PeekMessage(DataConstants.INPUT_PARAS, out msg);
                    resultMsg = string.Empty;

                    /* See what benchmark is to be performed */
                    for (int i = 0; i < 3; i++)
                        blobResult[i] = 0;
                    for (int i = 0; i < 20; i++)
                        tableResult[i] = 0;
                    for (int i = 0; i < 15; i++)
                        queueResult[i] = 0;

                    if (msg.AsString == "blob")
                    {
                        double[] temp = new double[10];
                        for (int cntr = 0; cntr < 5; cntr++)
                        {
                            temp = doBlobBenchMark();
                            for (int rslt = 0; rslt < 10; rslt++)
                            {
                                blobResult[rslt] += temp[rslt];
                            }
                        }

                        for (int i = 0; i < 10; i++)
                            resultMsg += blobResult[i].ToString() + ",";
                    }
                    else if (msg.AsString == "table")
                    {
                        double[] tempTable = new double[20];
                        tableResult = doTableBenchMark();
                        for (int i = 0; i < 20; i++)
                            resultMsg += tableResult[i].ToString() + ",";
                    }
                    else if (msg.AsString == "queue")
                    {
                        /*queueResult = doQueueBenchMark();
                        for (int i = 0; i < 15; i++)
                            resultMsg += queueResult[i].ToString() + ",";
                        */
                        queueResult = dothinkTimeBenchMark();
                        for (int i = 0; i < 15; i++)
                            resultMsg += queueResult[i].ToString() + ",";
                        for (int i = 0; i < 15; i++)
                            resultMsg += queueResult[i].ToString() + ",";

                    }

                    else if (msg.AsString == "tcp")
                    {
//                        queueResult = dotcpBenchMark();
                        for (int i = 0; i < 15; i++)
                            resultMsg += queueResult[i].ToString() + ",";
                        for (int i = 0; i < 15; i++)
                            resultMsg += queueResult[i].ToString() + ",";

                    }
                    else if (msg.AsString == "sql")
                    {
                        sqlResult = dosqlBenchMark();
                        for (int i = 0; i < 6; i++)
                            resultMsg += sqlResult[i].ToString() + ",";
                    }

                    while (!QueueObject.PutMessage(DataConstants.RESULT_QUEUE, new CloudQueueMessage(resultMsg)))
                    {
                        Thread.Sleep(1000);
                    }

                    int doneworkers = 0;
                    int totalWorkers = RoleEnvironment.Roles["WorkerRole"].Instances.Count;

                    while (doneworkers < totalWorkers)
                    {
                        QueueObject.GetApproxMessageCount(DataConstants.RESULT_QUEUE, out doneworkers);
                        Thread.Sleep(1000);
                    }
                    while (!QueueObject.ClearMessages(DataConstants.INPUT_PARAS))
                    {
                        Thread.Sleep(1000);
                    }
                }

            }
            catch (Exception ex)
            {
                System.Diagnostics.Trace.WriteLine(ex.Message);
            }
        }
        public double[] dosqlBenchMark()
        {
            try
            {
                double[] result = new double[6];
                int roleID = -1;
                getRoleID(out roleID);

                string sqlconnectionString = "Data Source=j5e8268vyc.database.windows.net;Initial Catalog=azurebenchmarks;Persist Security Info=True;User ID=dagarwal2;Password=NemeSis2007";
                SqlConnection conn = new SqlConnection(sqlconnectionString);
                conn.Open();
                if (conn.State != System.Data.ConnectionState.Open)
                {
                    System.Diagnostics.Trace.WriteLine("Connection to SQL failed. ", "Error");
                }
                else
                {
                    /* Drop the table if left from last run */
                    result[0] = dropSqlTable(conn, DataConstants.SQL_TABLE + roleID);
                    /* Create a table */
                    result[0] = createSqlTable(conn, DataConstants.SQL_TABLE + roleID);
                    /* Insert records */
                    result[1] = insertSql(conn, DataConstants.SQL_TABLE + roleID, DataConstants.SQL_RECORDS);
                    /* Update records */
                    result[2] = UpdateSql(conn, DataConstants.SQL_TABLE + roleID, DataConstants.SQL_RECORDS);
                    /* select records */
                    result[3] = SelectSql(conn, DataConstants.SQL_TABLE + roleID, DataConstants.SQL_RECORDS);
                    /* Delete records */
                    result[4] =  DeleteSql(conn, DataConstants.SQL_TABLE + roleID, DataConstants.SQL_RECORDS);
                    /* Drop Table */
                    result[5] = dropSqlTable(conn, DataConstants.SQL_TABLE + roleID);
                }

                return result;
            }
            catch (Exception ex)
            {
                System.Diagnostics.Trace.WriteLine("Error in SQL Benchmarks. " + ex.Message, "Error");
                throw(ex);
            }
        }

        public double[] doBlobBenchMark()
        {
            try
            {   
                double[] result = new double[10];
                double[] indiTime = new double[2];
                BlobHelper blobObject = new BlobHelper(connectionString);
                QueueHelper queueObject = new QueueHelper(connectionString);

                int roleID = 0;
                getRoleID(out roleID);
                /********************************************************/
                indiTime = uploadBlob();
                if (roleID == 0) flushBlob(DataConstants.BLOB_CONTAINER, DataConstants.BENCH_BLOCK_BLOB);
                result[1] = indiTime[1];
                result[0] = indiTime[0];
                /* Wait for worker 0 to finish */
                syncCount = syncCount + 1;
                synchronize(syncCount);

                System.Diagnostics.Trace.WriteLine("Done uploading blob - worker " + roleID);
                /**************************************************/
                indiTime = randomBlobAccess();
                result[8] = indiTime[0];
                result[9] = indiTime[1];
                /* Wait for everyone to download it before you delete it */
                syncCount = syncCount + 1;
                synchronize(syncCount);
                System.Diagnostics.Trace.WriteLine("Done random accessing blob - worker " + roleID);
                /*************************************************/
                indiTime = downloadBlob();
                result[2] = indiTime[0];
                result[3] = indiTime[1];
                /* Wait for everyone to download it before you delete it */
                syncCount = syncCount + 1;
                synchronize(syncCount);

                /***************Upload individual blobs***********************************/
                indiTime = uploadIndiBlob();
                result[4] = indiTime[0];
                result[5] = indiTime[1];
                //synchronize();
                /****************Download individual blobs**********************************/
                indiTime = downloadIndiBlob();
                result[6] = indiTime[0];
                result[7] = indiTime[1];
                //synchronize();
                while (!blobObject.DeletePageBlob(DataConstants.BLOB_CONTAINER, DataConstants.BENCH_PAGE_BLOB + roleID.ToString()))
                {
                    Thread.Sleep(1000);
                }
                while (!blobObject.DeleteBlob(DataConstants.BLOB_CONTAINER, DataConstants.BENCH_BLOCK_BLOB + roleID.ToString()))
                {
                    Thread.Sleep(1000);
                }
                /**************************************************/
                return result;
            }
            catch (Exception ex)
            {
                System.Diagnostics.Trace.WriteLine("Error in blob benchmarks" + ex.Message);
                return null;
            }

        }

        public double[] doTableBenchMark()
        {
            try
            {
                double[] result = new double[20]; /* 4 operations x 5 sizes */
                TimeSpan timeInsert = new TimeSpan(0, 0, 0);
                TimeSpan timeUpdate = new TimeSpan(0, 0, 0);
                TimeSpan timeQuery = new TimeSpan(0, 0, 0);
                TimeSpan timeDelete = new TimeSpan(0, 0, 0);

                DateTime counter = new DateTime();
                int numberOfWorkers = RoleEnvironment.Roles["WorkerRole"].Instances.Count;

                int entityCount = DataConstants.TABLE_COUNT;
                int entitySize = DataConstants.TABLE_ENTITY_SIZE;
                int cntr = 0;
                for (cntr = 1; cntr <= 5; cntr++)
                {
                    System.Diagnostics.Trace.WriteLine(cntr.ToString());
                    counter = DateTime.Now;
                    insertEntity(entityCount, entitySize);
                    timeInsert = DateTime.Now.Subtract(counter);

                    counter = DateTime.Now;
                    queryEntity(entityCount);
                    timeQuery = DateTime.Now.Subtract(counter);

                    counter = DateTime.Now;
                    updateEntity(entitySize, entityCount);
                    timeUpdate = DateTime.Now.Subtract(counter);

                    counter = DateTime.Now;
                    deleteEntity(entityCount);
                    timeDelete = DateTime.Now.Subtract(counter);

                    result[(cntr - 1) * 4] = timeInsert.TotalSeconds;
                    result[(cntr - 1) * 4 + 1] = timeQuery.TotalSeconds;
                    result[(cntr - 1) * 4 + 2] = timeUpdate.TotalSeconds;
                    result[(cntr - 1) * 4 + 3] = timeDelete.TotalSeconds;

                    entitySize *= 2;
                }

                return result;
            }
            catch (Exception ex)
            {
                System.Diagnostics.Trace.WriteLine("Error in doTableBenchmark. " + ex.Message, "Error");
                return null;
            }
        }

        public double[] doQueueBenchMark()
        {
            try
            {
                double[] result = new double[15];
                TimeSpan timePut = new TimeSpan(0, 0, 0);
                TimeSpan timePeek = new TimeSpan(0, 0, 0);
                TimeSpan timeGet = new TimeSpan(0, 0, 0);
                DateTime counter = new DateTime();
                int numberOfWorkers = RoleEnvironment.Roles["WorkerRole"].Instances.Count;
                QueueHelper queueObject = new QueueHelper(connectionString);
                int roleID = 0;
                getRoleID(out roleID);
                queueObject.CreateQueue(DataConstants.BENCH_QUEUE + roleID.ToString());
                int msgCount = 0;
                int msgSize = DataConstants.MSG_SIZE;
                msgCount = DataConstants.MSG_COUNT_LARGE / numberOfWorkers;
                for (int cntr = 1; cntr <= 5; cntr++)
                {
                    timePut = putMessages(DataConstants.BENCH_QUEUE + roleID.ToString(), msgSize, msgCount);

                    counter = DateTime.Now;
                    peekMessages(DataConstants.BENCH_QUEUE + roleID.ToString(), msgCount);
                    timePeek = DateTime.Now.Subtract(counter);

                    counter = DateTime.Now;
                    getMessages(DataConstants.BENCH_QUEUE + roleID.ToString(), msgCount);
                    timeGet = DateTime.Now.Subtract(counter);

                    result[(cntr - 1) * 3] = timePut.TotalSeconds;
                    result[(cntr - 1) * 3 + 1] = timePeek.TotalSeconds;
                    result[(cntr - 1) * 3 + 2] = timeGet.TotalSeconds;

                    msgSize *= 2;
                }
                queueObject.DeleteQueue(DataConstants.BENCH_QUEUE + roleID.ToString());
                return result;
            }
            catch (Exception ex)
            {
                System.Diagnostics.Trace.WriteLine("Error: " + ex.Message);
                throw;
            }
        }

        public double[] dothinkTimeBenchMark()
        {
            try
            {
                double[] result = new double[15];
                TimeSpan timePut = new TimeSpan(0, 0, 0);
                TimeSpan timePeek = new TimeSpan(0, 0, 0);
                TimeSpan timeGet = new TimeSpan(0, 0, 0);
                DateTime counter = new DateTime();
                int numberOfWorkers = RoleEnvironment.Roles["WorkerRole"].Instances.Count;
                QueueHelper queueObject = new QueueHelper(connectionString);
                int roleID = 0;
                int msgSize = 12 * 1024;
                int msgCount = DataConstants.QUEUE_BW_LIMIT / numberOfWorkers;
                int totalRounds = DataConstants.MSG_COUNT_LARGE / DataConstants.QUEUE_BW_LIMIT;
                getRoleID(out roleID);

                /* We write 500 messages at a time but we write to the same queue */
                int thinkTime = DataConstants.THINK_TIME;

                for (int think = 0; think < 5; think++)
                {
                    for (int cntr = 1; cntr <= totalRounds; cntr++)
                    {
                        timePut = putMessages(DataConstants.BENCH_QUEUE, msgSize, msgCount);

                        Thread.Sleep(thinkTime);

                        counter = DateTime.Now;
                        peekMessages(DataConstants.BENCH_QUEUE, msgCount);
                        timePeek = DateTime.Now.Subtract(counter);

                        Thread.Sleep(thinkTime);

                        counter = DateTime.Now;
                        getMessages(DataConstants.BENCH_QUEUE, msgCount);
                        timeGet = DateTime.Now.Subtract(counter);

                        Thread.Sleep(thinkTime);

                        result[think * 3] += timePut.TotalSeconds;
                        result[think * 3 + 1] += timePeek.TotalSeconds;
                        result[think * 3 + 2] += timeGet.TotalSeconds;

                    }
                    thinkTime = thinkTime + 1000 ;
                }
                return result;
            }
            catch (Exception ex)
            {
                System.Diagnostics.Trace.WriteLine("Error: " + ex.Message);
                throw;
            }
        }

        public double[] uploadBlob()
        {
            try
            {
                double[] result = new double[2];
                TimeSpan timePage = new TimeSpan(0, 0, 0);
                TimeSpan timeBlock = new TimeSpan(0, 0, 0);
                DateTime counter = new DateTime();
                int roleID = 0;
                BlobHelper blobObject = new BlobHelper(connectionString);

                getRoleID(out roleID);
                /*******************************************************************/
                /* Upload the page blob. Every worker uploads a separate page blob of size BLOB_SIZE */
                System.Diagnostics.Trace.WriteLine("Started page blob uploading at " + DateTime.Now.ToString() + ".");
                counter = DateTime.Now;
                blobObject.PutRandomPage(DataConstants.BLOB_CONTAINER, DataConstants.BENCH_PAGE_BLOB);
                System.Diagnostics.Trace.WriteLine("Page blob uploading ends at " + DateTime.Now.ToString() + ".");
                timePage = DateTime.Now.Subtract(counter);

                /******************************************************************/
                string[] blockIDs = new string[DataConstants.BLOB_SIZE / DataConstants.BLOCK_SIZE]; /* Required to download blockIDs to commit a block */

                /* upload a block blob - every worker uploads one blob.*/
                System.Diagnostics.Trace.WriteLine("Started block blob uploading at " + DateTime.Now.ToString() + ".");

                int from = 0;
                int to = 0;
                int RoleCount = RoleEnvironment.CurrentRoleInstance.Role.Instances.Count();
                int numBlocks = DataConstants.BLOB_SIZE / DataConstants.BLOCK_SIZE;
                int remainder = numBlocks % RoleCount;

                from = roleID * ((int)Math.Floor((double)numBlocks / RoleCount)) + Math.Min(roleID, remainder);
                to = (roleID + 1) * ((int)Math.Floor((double)numBlocks / RoleCount)) + Math.Min(roleID + 1, remainder) - 1;

                counter = DateTime.Now;
                byte[] blob = GenerateBytes(DataConstants.BLOCK_SIZE);
                string blockID;
                for (int i = from; i <= to; i++)
                {
                    blockID = Guid.NewGuid().ToString();
                    BlobHelper.PutBlock(connectionString, DataConstants.BLOB_CONTAINER, DataConstants.BENCH_BLOCK_BLOB, blockID, blob);
                }
                timeBlock = DateTime.Now.Subtract(counter);
                System.Diagnostics.Trace.WriteLine("Block blob uploading ends at " + DateTime.Now.ToString() + ".");
                /***************************************************************/
                result[0] = timePage.TotalSeconds;
                result[1] = timeBlock.TotalSeconds;

                while (!QueueHelper.PutMessageStatic(connectionString, DataConstants.WORKER_TERMINATION_INDICATOR, new CloudQueueMessage("uploaded" + roleID.ToString())))
                {
                    Thread.Sleep(1000);
                }

                return result;
            }
            catch (Exception ex)
            {
                System.Diagnostics.Trace.WriteLine("Error in uploading blob" + ex.Message);
                return null;
            }
        }

        public double[] downloadBlob()
        {
            try
            {
                double[] result = new double[2];
                TimeSpan timePage = new TimeSpan(0, 0, 0);
                TimeSpan timeBlock = new TimeSpan(0, 0, 0);
                DateTime counter = new DateTime();
                int roleID;
                getRoleID(out roleID);

                BlobHelper blobObject = new BlobHelper(connectionString);
                /*******************************************************************/
                string blob = string.Empty;
                counter = DateTime.Now;
                blobObject.GetPage(DataConstants.BLOB_CONTAINER, DataConstants.BENCH_PAGE_BLOB, 0, DataConstants.BLOB_SIZE, out blob);
                timePage = DateTime.Now.Subtract(counter);
                blob = string.Empty;
                /******************************************************************/
                /* download the block blob - every worker downloads one blob.*/
                counter = DateTime.Now;
                blobObject.GetBlob(DataConstants.BLOB_CONTAINER, DataConstants.BENCH_BLOCK_BLOB, out blob);
                timeBlock = DateTime.Now.Subtract(counter);
                blob = string.Empty;
                /***************************************************************/

                result[0] = timePage.TotalSeconds;
                result[1] = timeBlock.TotalSeconds;
                while (!QueueHelper.PutMessageStatic(connectionString, DataConstants.WORKER_TERMINATION_INDICATOR, new CloudQueueMessage("downloaded" + roleID.ToString())))
                {
                    Thread.Sleep(1000);
                }
                return result;
            }
            catch (Exception ex)
            {
                System.Diagnostics.Trace.WriteLine("Error in Downloading blob" + ex.Message);
                return null;
            }
        }

        public double[] uploadIndiBlob()
        {
            try
            {
                double[] result = new double[2];
                TimeSpan timePage = new TimeSpan(0, 0, 0);
                TimeSpan timeBlock = new TimeSpan(0, 0, 0);
                DateTime counter = new DateTime();
                string[] blockIDs = new string[DataConstants.BLOB_SIZE / DataConstants.BLOCK_SIZE];

                int roleID = 0;
                getRoleID(out roleID);

                BlobHelper blobObject = new BlobHelper(connectionString);
                /*******************************************************************/
                System.Diagnostics.Trace.WriteLine("Started page blob uploading at " + roleID.ToString() + DateTime.Now.ToString() + ".");
                counter = DateTime.Now;

                while (!blobObject.PutBlob(DataConstants.BLOB_CONTAINER, DataConstants.BENCH_PAGE_BLOB + roleID.ToString(), DataConstants.BLOB_SIZE))
                {
                    Thread.Sleep(1000);
                }

                byte[] content = new byte[DataConstants.SINGLE_PAGE_SIZE];
                content = GenerateBytes(DataConstants.SINGLE_PAGE_SIZE);
                int pageOffset = 0;

                while (pageOffset < DataConstants.BLOB_SIZE)
                {
                    while (!blobObject.PutPage(DataConstants.BLOB_CONTAINER, DataConstants.BENCH_PAGE_BLOB + roleID.ToString(), content, pageOffset, DataConstants.SINGLE_PAGE_SIZE))
                    {
                        Thread.Sleep(1000);
                    }
                    pageOffset += DataConstants.SINGLE_PAGE_SIZE;
                }

                System.Diagnostics.Trace.WriteLine("Page blob uploading ends at " + roleID.ToString() + DateTime.Now.ToString() + ".");
                timePage = DateTime.Now.Subtract(counter);

                /******************************************************************/
                /* upload a block blob - every worker uploads one blob.*/
                string blockID;
                System.Diagnostics.Trace.WriteLine("Started block blob uploading at " + roleID.ToString() + DateTime.Now.ToString() + ".");
                for (int i = 0; i < (DataConstants.BLOB_SIZE / DataConstants.BLOCK_SIZE); i++)
                {
                    byte[] blob = GenerateBytes(DataConstants.BLOCK_SIZE);
                    counter = DateTime.Now;
                    blockID = Guid.NewGuid().ToString();
                    while (!BlobHelper.PutBlock(connectionString, DataConstants.BLOB_CONTAINER, DataConstants.BENCH_BLOCK_BLOB + roleID.ToString(), blockID, blob))
                    {
                        Thread.Sleep(1000);
                    }
                    timeBlock = timeBlock.Add(DateTime.Now.Subtract(counter));
                }

                flushBlob(DataConstants.BLOB_CONTAINER, DataConstants.BENCH_BLOCK_BLOB + roleID.ToString());
                System.Diagnostics.Trace.WriteLine("Page block uploading done at " + roleID.ToString() + DateTime.Now.ToString() + ".");
                /***************************************************************/
                result[0] = timePage.TotalSeconds;
                result[1] = timeBlock.TotalSeconds;

                return result;
            }
            catch (Exception ex)
            {
                System.Diagnostics.Trace.WriteLine("Error in uploading blob" + ex.Message);
                return null;
            }
        }

        public double[] downloadIndiBlob()
        {
            try
            {
                double[] result = new double[2];
                TimeSpan timePage = new TimeSpan(0, 0, 0);
                TimeSpan timeBlock = new TimeSpan(0, 0, 0);
                DateTime counter = new DateTime();

                string instanceId = RoleEnvironment.CurrentRoleInstance.Id;
                int roleID = 0;
                getRoleID(out roleID);

                BlobHelper blobObject = new BlobHelper(connectionString);
                /*******************************************************************/
                string blob = string.Empty;
                System.Diagnostics.Trace.WriteLine("Started page blob downloading at " + DateTime.Now.ToString() + ".");
                counter = DateTime.Now;
                while (!blobObject.GetPage(DataConstants.BLOB_CONTAINER, DataConstants.BENCH_PAGE_BLOB + roleID.ToString(), 0, DataConstants.BLOB_SIZE, out blob))
                {
                    Thread.Sleep(1000);
                }
                System.Diagnostics.Trace.WriteLine("Page blob downloading ends at " + DateTime.Now.ToString() + ".");
                timePage = DateTime.Now.Subtract(counter);
                blob = string.Empty;

                /******************************************************************/
                /* download the block blob - every worker downloads one blob.*/
                counter = DateTime.Now;
                System.Diagnostics.Trace.WriteLine("Started block blob downloading at " + DateTime.Now.ToString() + ".");
                //blobObject.BlobStreamDownload(DataConstants.BLOB_CONTAINER, DataConstants.BENCH_BLOCK_BLOB + roleID.ToString());
                MemoryStream blobContent = new MemoryStream();
                blobObject.GetBlob(DataConstants.BLOB_CONTAINER, DataConstants.BENCH_BLOCK_BLOB + roleID.ToString(), out blobContent);
                blobContent.Close();
                System.Diagnostics.Trace.WriteLine("Block blob downloading ends at " + DateTime.Now.ToString() + ".");
                timeBlock = DateTime.Now.Subtract(counter);
                blob = string.Empty;
                /***************************************************************/
                result[0] = timePage.TotalSeconds;
                result[1] = timeBlock.TotalSeconds;
                //QueueHelper.PutMessageStatic(connectionString, DataConstants.WORKER_TERMINATION_INDICATOR, new CloudQueueMessage("downloaded"));

                return result;
            }
            catch (Exception ex)
            {
                System.Diagnostics.Trace.WriteLine("Error in Downloading blob" + ex.Message);
                return null;
            }
        }

        public double[] randomBlobAccess()
        {
            try
            {
                double[] result = new double[2];
                TimeSpan timePage = new TimeSpan(0, 0, 0);
                TimeSpan timeBlock = new TimeSpan(0, 0, 0);
                DateTime counter = new DateTime();
                Random getRandomNumber = new Random();

                int roleID = 0;
                getRoleID(out roleID);

                BlobHelper blobObject = new BlobHelper(connectionString);
                /*******************************************************************/
                /* Read random pages from the Page Blob*/
                string blob = string.Empty;
                System.Diagnostics.Trace.WriteLine("Started page blob random access at " + DateTime.Now.ToString() + ".");
                int totalPages = (DataConstants.BLOB_SIZE / DataConstants.SINGLE_PAGE_SIZE);
                counter = DateTime.Now;
                for (int i = 0; i < totalPages; i++)
                {
                    int pageOffset = getRandomNumber.Next(totalPages);
                    while (!blobObject.GetPage(DataConstants.BLOB_CONTAINER, DataConstants.BENCH_PAGE_BLOB, pageOffset, DataConstants.SINGLE_PAGE_SIZE, out blob))
                    {
                        Thread.Sleep(1000);
                    }
                }
                timePage = DateTime.Now.Subtract(counter);
                System.Diagnostics.Trace.WriteLine("Page blob random access ends at " + DateTime.Now.ToString() + ".");
                blob = string.Empty;

                /******************************************************************/
                /* Read blocks from the Block Blob. Since Blobs don't support random access we will read sequentially block by block*/
                counter = DateTime.Now;
                System.Diagnostics.Trace.WriteLine("Started block blob stream downloading at " + DateTime.Now.ToString() + ".");
                blobObject.BlobStreamDownload(DataConstants.BLOB_CONTAINER, DataConstants.BENCH_BLOCK_BLOB);
                System.Diagnostics.Trace.WriteLine("Block blob stream downloading ends at " + DateTime.Now.ToString() + ".");
                timeBlock = DateTime.Now.Subtract(counter);
                blob = string.Empty;
                /***************************************************************/
                result[0] = timePage.TotalSeconds;
                result[1] = timeBlock.TotalSeconds;
                while (!QueueHelper.PutMessageStatic(connectionString, DataConstants.WORKER_TERMINATION_INDICATOR, new CloudQueueMessage("randomaccess" + roleID.ToString())))
                {
                    Thread.Sleep(1000);
                }

                return result;
            }
            catch (Exception ex)
            {
                System.Diagnostics.Trace.WriteLine("Error in Downloading blob" + ex.Message);
                return null;
            }

        }

        public void getRoleID(out int roleID)
        {
            string instanceId = RoleEnvironment.CurrentRoleInstance.Id;
            if (!int.TryParse(instanceId.Substring(instanceId.LastIndexOf(".") + 1), out roleID)) // On cloud.
            {
                int.TryParse(instanceId.Substring(instanceId.LastIndexOf("_") + 1), out roleID); // On compute emulator.
            }
        }

        public void synchronize(int syncCount)
        {
            QueueHelper queueObject = new QueueHelper(connectionString);
            int doneWorkers = 0;
            int totalWorkers = RoleEnvironment.CurrentRoleInstance.Role.Instances.Count();
            while (doneWorkers < (totalWorkers * syncCount) )
            {
                queueObject.GetApproxMessageCount(DataConstants.WORKER_TERMINATION_INDICATOR, out doneWorkers);
                Thread.Sleep(1000);
            }
        }

        public void flushBlob(string container, string blobname)
        {
            string[] blockIDs;
            BlobHelper.GetBlockList(connectionString, container, blobname, out blockIDs);
            BlobHelper.PutBlockList(connectionString, container, blobname, blockIDs);
        }

        public TimeSpan putMessages(string queuename, int msgSize, int msgCount)
        {
            QueueHelper queueObject = new QueueHelper(connectionString);
            int roleID = 0;
            getRoleID(out roleID);
            TimeSpan timeTaken = new TimeSpan(0, 0, 0);
            for (int msgNum = 0; msgNum < msgCount; msgNum++)
            {
                StringBuilder msg = GenerateRandomString(msgSize);
                DateTime start = DateTime.Now;
                while (!queueObject.PutMessage(queuename, new CloudQueueMessage(msg.ToString())))
                {
                    Thread.Sleep(1000);
                }
                timeTaken = timeTaken.Add(DateTime.Now.Subtract(start));
            }
            return timeTaken;
        }

        public void peekMessages(string queuename, int msgCount)
        {
            QueueHelper queueObject = new QueueHelper(connectionString);
            CloudQueueMessage msg = new CloudQueueMessage("");
            int roleID = 0;
            getRoleID(out roleID);

            for (int msgNum = 0; msgNum < msgCount; msgNum++)
            {
                while (!queueObject.PeekMessage(queuename, out msg))
                {
                    Thread.Sleep(1000);
                }
            }
        }

        public void getMessages(string queuename, int msgCount)
        {
            QueueHelper queueObject = new QueueHelper(connectionString);
            CloudQueueMessage msg = new CloudQueueMessage("");
            int roleID = 0;
            getRoleID(out roleID);

            for (int msgNum = 0; msgNum < msgCount; msgNum++)
            {
                while (!queueObject.GetMessage(queuename, out msg))
                {
                    Thread.Sleep(1000);
                }
                queueObject.DeleteMessage(queuename, msg);
            }
        }

        public void insertEntity(int entityCount, int entitySize)
        {
            try
            {
                TableHelper tableObject = new TableHelper(connectionString);
                StringBuilder entity_data = GenerateRandomString(entitySize);

                TableData entity = new TableData();
                entity.Data = entity_data.ToString();
                string instanceId = RoleEnvironment.CurrentRoleInstance.Id;
                int roleID = 0;
                if (!int.TryParse(instanceId.Substring(instanceId.LastIndexOf(".") + 1), out roleID)) // On cloud.
                {
                    int.TryParse(instanceId.Substring(instanceId.LastIndexOf("_") + 1), out roleID); // On compute emulator.
                }

                entity.PartitionKey = roleID.ToString();

                for (int i = 0; i < entityCount; i++)
                {
                    entity.RowKey = i.ToString();
                    while (!tableObject.InsertEntity(DataConstants.BENCH_TABLE, entity))
                    {
                        Thread.Sleep(1000);
                    }
                }
            }
            catch (Exception ex)
            {
                System.Diagnostics.Trace.WriteLine("Error in insert entity. " + ex.Message, "Error");
            }

        }

        public void queryEntity(int entityCount)
        {
            try
            {
                TableHelper tableObject = new TableHelper(connectionString);

                /* Every worker should query the same entity */
                string rowKey = "0";
                string instanceId = RoleEnvironment.CurrentRoleInstance.Id;
                int roleID = 0;
                if (!int.TryParse(instanceId.Substring(instanceId.LastIndexOf(".") + 1), out roleID)) // On cloud.
                {
                    int.TryParse(instanceId.Substring(instanceId.LastIndexOf("_") + 1), out roleID); // On compute emulator.
                }

                string partitionKey = roleID.ToString();
                string data;
                TableData entity = new TableData();
                for (int i = 0; i < entityCount; i++)
                {
                    if (tableObject.GetEntity(DataConstants.BENCH_TABLE, partitionKey, rowKey, out entity))
                    {
                        data = entity.Data;
                    }
                }
            }
            catch (Exception ex)
            {
                System.Diagnostics.Trace.WriteLine("Error in Query entity. " + ex.Message, "Error");
            }
        }

        public void deleteEntity(int entityCount)
        {
            try
            {
                TableHelper tableObject = new TableHelper(connectionString);
                TableData entity = new TableData();
                string data = string.Empty;
                string instanceId = RoleEnvironment.CurrentRoleInstance.Id;
                int roleID = 0;
                if (!int.TryParse(instanceId.Substring(instanceId.LastIndexOf(".") + 1), out roleID)) // On cloud.
                {
                    int.TryParse(instanceId.Substring(instanceId.LastIndexOf("_") + 1), out roleID); // On compute emulator.
                }

                string partitionKey = roleID.ToString();
                for (int i = 0; i < entityCount; i++)
                {
                    tableObject.DeleteEntity<TableData>(DataConstants.BENCH_TABLE, partitionKey, i.ToString());
                }
                //tableObject.DeleteTable(DataConstants.BENCH_TABLE);
            }
            catch (Exception ex)
            {
                System.Diagnostics.Trace.WriteLine("Error in delete Query. " + ex.Message, "Error");
            }
        }

        public void updateEntity(int entitySize, int entityCount)
        {
            try
            {
                TableHelper tableObject = new TableHelper(connectionString);
                StringBuilder newData = GenerateRandomString(entitySize);
                string instanceId = RoleEnvironment.CurrentRoleInstance.Id;
                int roleID = 0;
                if (!int.TryParse(instanceId.Substring(instanceId.LastIndexOf(".") + 1), out roleID)) // On cloud.
                {
                    int.TryParse(instanceId.Substring(instanceId.LastIndexOf("_") + 1), out roleID); // On compute emulator.
                }

                string partitionKey = roleID.ToString();

                for (int i = 0; i < entityCount; i++)
                {
                    tableObject.UnconditionalUpdate<TableData>(DataConstants.BENCH_TABLE, partitionKey, i.ToString(), newData.ToString());
                }
            }
            catch (Exception ex)
            {
                System.Diagnostics.Trace.WriteLine("Error in update Query. " + ex.Message, "Error");
            }
        }

        public void dotcpbenchmark()
        {
            int roleID = 0;
            getRoleID(out roleID);
            if (roleID % 2 == 0)
            {
                

            }
        }
        public double[] readTCP(int msgSize)
        {
            double[] time = new double[2];
            TimeSpan timeTaken = new TimeSpan(0,0,0);
            TcpListener listener = startListener(RoleEnvironment.CurrentRoleInstance.InstanceEndpoints["tcp"].IPEndpoint);
            /* Keep waiting for a connection */
            while (!listener.Pending())
            {
                Thread.Sleep(1000);
            }

            /* Accept the connection */
            TcpClient reader = listener.AcceptTcpClient();
            NetworkStream readRequest = reader.GetStream();



            return time;
        }

        private TcpListener startListener(IPEndPoint endPt)
        {
            TcpListener listener = new TcpListener(endPt);
            listener.ExclusiveAddressUse = false;
            listener.Start();
            return listener;
        }


        private double createSqlTable(SqlConnection conn, string tableName)
        {
            TimeSpan timeTaken = new TimeSpan(0, 0, 0);
            DateTime start = DateTime.Now;
            using (SqlCommand command = new SqlCommand("CREATE TABLE "+ tableName + " (rowid int NOT NULL, txtdata varchar(4000), CONSTRAINT [PK_Source] PRIMARY KEY CLUSTERED ([rowid]))", conn))
            {
                command.ExecuteNonQuery();
            }
            timeTaken = timeTaken.Add(DateTime.Now.Subtract(start));
            return timeTaken.TotalSeconds;
        }

        private double dropSqlTable(SqlConnection conn, string tableName)
        {
            TimeSpan timeTaken = new TimeSpan(0, 0, 0);
            DateTime start = DateTime.Now;
            using (SqlCommand command = new SqlCommand("IF EXISTS (SELECT * FROM sys.tables where name ='" + tableName + "') DROP TABLE " + tableName, conn))
            {
                command.ExecuteNonQuery();
            }
            timeTaken = timeTaken.Add(DateTime.Now.Subtract(start));
            return timeTaken.TotalSeconds;
        }

        private double insertSql(SqlConnection conn, string tablename, int recordCount)
        {
            TimeSpan timeTaken = new TimeSpan(0, 0, 0);
            DateTime start = DateTime.Now;
            int id = 0;
            StringBuilder data;
            try
            {
                data = GenerateRandomString(DataConstants.SQL_DATA_SIZE);
                for (int i = 0; i < recordCount; i++)
                {
                    id = id + 1;
                    string query = "insert into " + tablename + "(rowid,txtdata) VALUES(" + id.ToString() + ",'" + data.ToString() + "')";
                    using (SqlCommand command = new SqlCommand(query, conn))
                    {
                        command.Parameters.Add(new SqlParameter("rowid", id.ToString()));
                        command.Parameters.Add(new SqlParameter("txtdata", data.ToString()));
                        command.ExecuteNonQuery();
                    }
                }
                timeTaken = timeTaken.Add(DateTime.Now.Subtract(start));
                return timeTaken.TotalSeconds;
            }
            catch (Exception ex)
            {
                throw (ex);
            }
        }

        private double UpdateSql(SqlConnection conn, string tablename, int recordCount)
        {
            TimeSpan timeTaken = new TimeSpan(0, 0, 0);
            DateTime start = DateTime.Now;
            int id = 0;
            StringBuilder data;
            try
            {
                data = GenerateRandomString(DataConstants.SQL_DATA_SIZE);
                for (int i = 0; i < recordCount; i++)
                {
                    id = id + 1;
                    using (SqlCommand command = new SqlCommand("update " + tablename + " SET txtdata = '" + data.ToString() + "' WHERE (rowid = " + id.ToString() + ")", conn))
                    {
                        command.ExecuteNonQuery();
                    }
                }
                timeTaken = timeTaken.Add(DateTime.Now.Subtract(start));
                return timeTaken.TotalSeconds;
            }
            catch (Exception ex)
            {
                throw (ex);
            }
        }

        private double DeleteSql(SqlConnection conn, string tablename, int recordCount)
        {
            TimeSpan timeTaken = new TimeSpan(0, 0, 0);
            DateTime start = DateTime.Now;
            int id = 0;
            StringBuilder data;
            try
            {
                data = GenerateRandomString(DataConstants.SQL_DATA_SIZE);
                for (int i = 0; i < recordCount; i++)
                {
                    id = id + 1;
                    using (SqlCommand command = new SqlCommand("delete from " + tablename  + " WHERE (rowid = " + id.ToString() + ")", conn))
                    {
                        command.ExecuteNonQuery();
                    }
                }
                timeTaken = timeTaken.Add(DateTime.Now.Subtract(start));
                return timeTaken.TotalSeconds;
            }
            catch (Exception ex)
            {
                throw (ex);
            }
        }
        
        private double SelectSql(SqlConnection conn, string tablename, int recordCount)
        {
            TimeSpan timeTaken = new TimeSpan(0, 0, 0);
            DateTime start = DateTime.Now;
            int id = 0;
            string data;
            try
            {
                for (int i = 0; i < recordCount; i++)
                {
                    id = id + 1;
                    using (SqlCommand command = new SqlCommand("select txtdata from " + tablename + " WHERE (rowid = " + id.ToString() + ")", conn))
                    {
                        SqlDataReader rdr = command.ExecuteReader();
                        while (rdr.Read())
                        {
                            data = rdr[0].ToString();
                        }
                        rdr.Close();
                    }
                }
                timeTaken = timeTaken.Add(DateTime.Now.Subtract(start));
                return timeTaken.TotalSeconds;
            }
            catch (Exception ex)
            {
                throw (ex);
            }
        }

        private StringBuilder GenerateRandomString(int size)
        {
            StringBuilder builder = new StringBuilder();
            Random random = new Random();
            char ch;
            for (int i = 0; i < size - 10; i++) /* C# adds 20 bytes to each string */
            {
                ch = Convert.ToChar(Convert.ToInt32(Math.Floor(26 * random.NextDouble() + 65)));
                builder.Append(ch);
            }
            return builder;
        }

        private byte[] GenerateBytes(int numberBytes)
        {
            byte[] data = new byte[numberBytes];
            Random rnd = new Random();
            rnd.NextBytes(data);
            return data;
        }

        public override bool OnStart()
        {
            // Set the maximum number of concurrent connections 
            ServicePointManager.DefaultConnectionLimit = 48;

            this.instance_id = RoleEnvironment.CurrentRoleInstance.Id.ToString();

            // For information ons handling configuration changes
            // see the MSDN topic at http://go.microsoft.com/fwlink/?LinkId=166357.
            CloudStorageAccount.SetConfigurationSettingPublisher((configName, configSetter) =>
            {
                // Provide the configSetter with the initial value
                configSetter(RoleEnvironment.GetConfigurationSettingValue(configName));

                RoleEnvironment.Changed += (sender, arg) =>
                {
                    foreach (var change in arg.Changes)
                    {
                        var configChange = change as RoleEnvironmentConfigurationSettingChange;
                        if (configChange != null)
                        {
                            RoleEnvironment.GetConfigurationSettingValue(
                                configChange.ConfigurationSettingName);
                        }
                    }
                };
            });


            var diag = DiagnosticMonitor.GetDefaultInitialConfiguration();
            diag.Logs.ScheduledTransferLogLevelFilter = LogLevel.Verbose;
            diag.Logs.ScheduledTransferPeriod = TimeSpan.FromMinutes(1.0);
            diag.WindowsEventLog.DataSources.Add("Application!*");
            diag.WindowsEventLog.DataSources.Add("System!*");
            diag.WindowsEventLog.ScheduledTransferPeriod = TimeSpan.FromMinutes(1.0);
            diag.DiagnosticInfrastructureLogs.ScheduledTransferPeriod = TimeSpan.FromMinutes(1.0);
            DiagnosticMonitor.Start("Microsoft.WindowsAzure.Plugins.Diagnostics.ConnectionString", diag);

            return base.OnStart();
        }

    }
}
